package com.jwater.core.manager.worker;

import com.jwater.core.common.IpPort;
import com.jwater.core.common.S;
import com.jwater.core.common.param.ParamMap;
import com.jwater.core.manager.common.DataKeys;
import com.jwater.core.manager.common.MsgKeys;
import com.jwater.core.manager.common.conf.WorkerConfParse;
import com.jwater.core.manager.common.model.ReportInfo;
import com.jwater.core.manager.common.model.WorkerInfo;
import com.jwater.core.manager.common.service.JwaterService;
import com.jwater.core.manager.worker.assign.AssignPort;
import com.jwater.core.manager.worker.assign.AssignProp;
import com.jwater.core.manager.worker.assign.WorkerEnv;
import com.jwater.core.manager.worker.report.ReportMetric;
import com.jwater.core.metrics.JvmMetric;
import com.jwater.core.net.Node;
import com.jwater.core.net.NodeHandler;
import com.jwater.core.net.SimpleNode;
import com.jwater.core.serialize.ProtoStuffSerialize;
import com.jwater.core.serialize.SerializeUtils;
import com.jwater.launch.DeployPathUtils;
import com.jwater.launch.LoaderUtils;
import jwater.org.springframework.boot.loader.JWaterSpringbootLauncher;
import jwater.org.springframework.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @description:worker节点
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class WorkerNodeImpl implements WorkerNode, NodeHandler {
    protected static Logger logger = LoggerFactory.getLogger(WorkerNodeImpl.class);

    // 主机
    private String host;
    // master的主机信息
    private List<IpPort> masterHosts;
    // worker信息
    protected WorkerInfo workerInfo;
    // worker编号
    protected int index;
    // client通信
    protected Node node;
    // rpc调用master服务
    private volatile JwaterService jwaterService;
    // 是否登录
    private volatile boolean isLogin = false;
    // 是否第1次启动
    private volatile boolean isFirst = true;
    // worker业务执行线程
    private volatile Thread workerThread;
    // 启动时间
    private long startTime = System.currentTimeMillis();
    // 上报状态
    private ReportMetric reportMetric;
    // handler
    private WorkerHandler handler;

    public WorkerNodeImpl(String host, List<IpPort> masterHosts, WorkerInfo workerInfo, int index) {
        this.host = host;
        this.masterHosts = masterHosts;
        this.workerInfo = workerInfo;
        this.index = index;
        this.workerInfo.setIndex(index);
        ((WorkerContextImpl) WorkerFactory.getContext()).workerNode = this;
        String key = DataKeys.getRunWorkerKey(workerInfo.getRegionName(), workerInfo.getDeployName(), workerInfo.getGroupName(),
                workerInfo.getWorkerName(), index);
        reportMetric = new ReportMetric(this, this.host, this.startTime, key);
    }

    @Override
    public void start() {
        node = new SimpleNode();
        node.setHandler(this);
        node.startAsClient(masterHosts, false);
        jwaterService = node.getProxy(JwaterService.class);
    }

    public void setHandler(WorkerHandler handler) {
        this.handler = handler;
    }

    private void checkService() {
        while (jwaterService == null) {
            try {
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                logger.error(e.toString(), e);
            }
        }
    }

    @Override
    public void onConnect(Node node) {
        try {
            if (isFirst) {
                isFirst = false;
                // 获取worker完整配置
                checkService();
                WorkerInfo workerConf = jwaterService.getWorkerInfo(workerInfo.getRegionName(), workerInfo.getDeployName(),
                        workerInfo.getGroupName(), workerInfo.getWorkerName());
                if (workerConf == null) {
                    logger.error("can not get worker conf,exit");
                    System.exit(0);
                    return;
                }
                this.workerInfo = workerConf;
                this.workerInfo.setIndex(this.index);
                logger.info("workerConf:" + workerConf);
                AssignPort.assign(workerInfo);
                AssignProp.assign(workerInfo);
            }
            // 登录
            byte[] bytes = SerializeUtils.serializeArray(host, workerInfo.getRegionName(), workerInfo.getDeployName(),
                    workerInfo.getGroupName(), workerInfo.getWorkerName(), index, startTime,
                    JvmMetric.pid(), WorkerEnv.getInstance().envsToString());
            bytes = node.syncCall(MsgKeys.LOGIN, bytes);
            isLogin = "true".equals(new String(bytes, S.charset));
            logger.info("login:" + isLogin);
            if (!isLogin) {
                logger.error("login fail,exit");
                System.exit(0);
                return;
            }
            // 运行主进程
            if (workerThread == null) {
                synchronized (this) {
                    if (workerThread == null) {
                        workerThread = new Thread(this::runWorker);
                        workerThread.setDaemon(false);
                        workerThread.start();
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        } finally {
            if (handler != null) {
                handler.onConnect(node);
            }
        }
    }

    @Override
    public void onDisconnect(Node node) {
        logger.info("onDisconnect:" + node.getNodeId());
        if (handler != null) {
            handler.onDisconnect(node);
        }
    }

    @Override
    public void onData(Node node, String dataType, byte[] data) {
        logger.info("onData:" + dataType);
        if (handler != null) {
            handler.onData(node, dataType, data);
        }
    }

    @Override
    public byte[] onSyncCall(Node node, String method, byte[] args) {
        logger.info("onSyncCall:" + method);
        if (MsgKeys.NOTIFY_EXIT.equals(method)) {
            if (handler != null) {
                handler.onExit();
            }
            return new byte[0];
        } else {
            if (handler != null) {
                return handler.onSyncCall(node, method, args);
            }
        }
        return new byte[0];
    }

    @Override
    public void onAsyncCall(Node node, String method, byte[] args, ReturnCallback callback) {
        logger.info("onAsyncCall:" + method);
        if (handler != null) {
            handler.onAsyncCall(node, method, args, callback);
        }
    }

    private void runWorker() {
        try {
            loadToClasspath(workerInfo);
            reportMetric.start();
            boolean springboot = "true".equals(workerInfo.getProps().getString("springboot", "false"));
            if (springboot) {
                logger.info("spring boot loader run:" + workerInfo.getClassName());
                new JWaterSpringbootLauncher(workerInfo.getClassName()).launch(getMainArgs(workerInfo));
            } else {
                logger.info("run:" + workerInfo.getClassName());
                LoaderUtils.exeMain(workerInfo.getClassName(), getMainArgs(workerInfo));
            }
        } catch (Throwable e) {
            logger.error(e.toString(), e);
        }
    }

    public void report(ReportInfo reportInfo) {
        if (node == null || !node.isConnect()) {
            return;
        }
        ProtoStuffSerialize protoStuffSerialize = new ProtoStuffSerialize();
        byte[] bytes = protoStuffSerialize.serialize(reportInfo);
        node.sendData(MsgKeys.REPORT, bytes);
    }

    private void loadToClasspath(WorkerInfo workerConf) {
        String regionName = workerConf.getRegionName();
        String deployName = workerConf.getDeployName();
        ParamMap props = workerConf.getProps();
        String jarDir = props.getString("jarDir");
        String confDir = props.getString("confDir");
        String classpathDir = props.getString("classpathDir");
        String include = props.getString("include");
        Long version = workerConf.getInfoProps().getLong("version");
        List<String> confPaths = new ArrayList<String>();
        List<String> jarPaths = new ArrayList<String>();
        File deployPath = DeployPathUtils.getDeployPath(regionName, deployName, version);
        if (deployPath != null && deployPath.exists()) {
            jarPaths.add(deployPath.getPath() + "/lib");
            if (!StringUtils.isEmpty(jarDir)) {
                for (String dir : jarDir.split(",")) {
                    jarPaths.add(deployPath.getPath() + "/" + dir);
                }
            }
            confPaths.add(deployPath.getPath() + "/conf");
            if (!StringUtils.isEmpty(confDir)) {
                for (String dir : confDir.split(",")) {
                    confPaths.add(deployPath.getPath() + "/" + dir);
                }
            }
            if (!StringUtils.isEmpty(classpathDir)) {
                for (String dir : classpathDir.split(",")) {
                    confPaths.add(deployPath.getPath() + "/" + dir);
                }
            }
        }
        if (!StringUtils.isEmpty(include)) {
            for (String dir : include.split(",")) {
                File includeDeployPath = WorkerConfParse.getDeployPath(regionName, dir);
                if (includeDeployPath != null && includeDeployPath.exists()) {
                    confPaths.add(includeDeployPath.getPath() + "/conf");
                    jarPaths.add(includeDeployPath.getPath() + "/lib");
                }
            }
        }
        try {
            for (String jarPath : jarPaths) {
                LoaderUtils.addJarDir(new File(jarPath), true);
            }
            for (String confPath : confPaths) {
                LoaderUtils.loadDirToClasspath(new File(confPath));
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }

        List<String> allJars = LoaderUtils.getAllJars();
        for (String allJar : allJars) {
            logger.info(allJar);
        }
    }

    private String[] getMainArgs(WorkerInfo workerConf) {
        String mainArgs = workerConf.getProps().getString("mainArgs");
        if (StringUtils.isEmpty(mainArgs)) {
            return new String[]{};
        }
        return mainArgs.split("\\s+");
    }
}
